package com.suxin.test.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.example.netty.adv.client.CilentBusiRespQueue;
import org.example.netty.adv.client.ClientInit;
import org.example.netty.adv.vo.EncryptUtils;
import org.example.netty.adv.vo.Message;
import org.example.netty.adv.vo.MessageHeader;
import org.example.netty.adv.vo.MessageType;

import java.net.InetSocketAddress;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * @Author: tang
 * @Description:
 * @Date 2025/1/21 15:28
 */
@Slf4j
public class BusiClient implements Runnable{

    private String host;

    private int port;

    private Channel channel;

    private volatile boolean isConnect = false;

    private volatile boolean userClose = false;

    private final NioEventLoopGroup bossGroup = new NioEventLoopGroup();

    public BusiClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    @Override
    public void run() {
        if (this.userClose) {
            return;
        }
        try {
            connect();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }finally {
            log.warn("client 重连服务开始....");
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            this.run();
        }
    }

    public void connect() throws Exception{
        Bootstrap bootstrap = new Bootstrap();
        ChannelFuture channelFuture = bootstrap
                .group(bossGroup)
                .channel(NioSocketChannel.class)
                .handler(new ClientInit())
                .connect(new InetSocketAddress(this.host,this.port));
        log.info("client runing...");
        this.channel = channelFuture.channel();

        synchronized (this) {
            this.isConnect = true;
            this.notifyAll();
        }

        this.channel.closeFuture().sync();
        log.info("client close...");
    }

    public boolean isConnect(){
        return this.isConnect;
    }

    /**
     * sendMsg
     * @param message
     */
    public void sendMsg(Message message) {
        MessageHeader messageHeader = message.getMessageHeader();
        messageHeader.setMd5(EncryptUtils.encryptObj(message.getBody()));
        messageHeader.setMsgId(UUID.randomUUID().toString());
        byte type = messageHeader.getType();
        if (type == MessageType.SERVICE_RESP.value()) {
            sendTwoWayMsg(message,(msg) -> {
                System.out.println("client callback handler accept server :" + msg);
            });
        }else {
            sendOneWayMsg(message);
        }
    }

    /**
     * 需应答的业务请求消息
     * @param message
     */
    private void sendTwoWayMsg(Message message, Consumer<Message> consumer) {
        if (this.channel.isActive()) {
            sendOneWayMsg(message);
            CilentBusiRespQueue.put(message);
            Message poll = CilentBusiRespQueue.poll(message.getMessageHeader().getMsgId());
            consumer.accept(poll);
            CilentBusiRespQueue.remove(message);
        }
    }

    /**
     * 无需应答的业务请求消息
     * @param message
     */
    private void sendOneWayMsg(Message message) {
        if (this.channel.isActive()) {
            this.channel.writeAndFlush(message);
        }
    }

    public void close() {
        this.userClose = true;
        this.channel.close();
        this.bossGroup.shutdownGracefully();
    }
}
